package cn.onesdream.app.api.websocket;

import cn.onesdream.app.api.service.Impl.ChatServiceImpl;
import cn.onesdream.app.api.service.Impl.UserServiceImpl;
import cn.onesdream.app.dao.entity.Chat;
import cn.onesdream.app.dao.entity.User;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * @author WangShijun
 * @date 2020/10/9 14:24
 */
@ServerEndpoint(value = "/websocket/{account}")
@Controller
@Slf4j
public class AppWebSocketEndpoint {
    /**
     * 静态变量，用来记录当前连接数，应该把它设计成线程安全的。
     */
    private static AtomicInteger onlineCount = new AtomicInteger();

    /**
     * 线程安全的HashMap对象,ConcurrentHashMap性能比较好
     */
    private static Map<String, Session> onlineSessionMap = new ConcurrentHashMap<>();


    private static ChatServiceImpl chatService;

    private static UserServiceImpl userService;

    @Autowired
    public void setChatService(ChatServiceImpl chatService) {
        this.chatService = chatService;
    }

    @Autowired
    public void setUserService(UserServiceImpl userService) {
        this.userService = userService;
    }


    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("account") String myAccount) {
        onlineSessionMap.put(myAccount, session);
        addOnlineCount();           //在线数加1
        log.info("有新连接加入！用户是" + myAccount + "    当前在线人数为" + getOnlineCount());
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose(Session session, @PathParam("account") String myAccount) {
        onlineSessionMap.remove(myAccount);
        subOnlineCount();
        log.info("有一连接关闭！账户是：" + myAccount + "   当前在线人数为" + getOnlineCount());
        try {
            session.close();
        } catch (IOException e) {
            log.info("关闭资源时出错！");
            e.printStackTrace();
        }
    }

    /**
     * 收到客户端消息后调用的方法
     */
    @OnMessage
    public void onMessage(String message, @PathParam("account") String myAccount, Session session) throws IOException {
        int typeIndex = message.indexOf("!");
        int toAccountIndex = message.indexOf(":");
        String msgType = message.substring(0, typeIndex);
        if (msgType == null || "".equals(msgType)){
            onClose(session, myAccount);
        } else if ("0".equals(msgType)){
            String toAccount = message.substring(typeIndex + 1, toAccountIndex);
            if (toAccount == null || "".equals(toAccount)){
                onClose(session, myAccount);
            } else {
                sendMessageByAccount(toAccount, myAccount + ":" + message.substring(toAccountIndex + 1), myAccount, session);
                log.info("来自客户端的消息(发至用户" + toAccount + "):" + message + "   用户是：" + myAccount);
            }
        } else if ("1".equals(msgType)){
            sendBatchToOnline("群发消息【" + myAccount + "】:" + message.substring(toAccountIndex + 1), myAccount);
            log.info("来自客户端的消息(群发所有在线用户):" + message + "   用户是：" + myAccount);
        } else if ("2".equals(msgType)){
            sendBatchToAll("群发消息【" + myAccount + "】:" + message.substring(toAccountIndex + 1), myAccount);
            log.info("来自客户端的消息(群发所有注册用户):" + message + "   用户是：" + myAccount);
        }

    }

    /**
     * 发生错误时调用
     */
    @OnError
    public void onError(@PathParam("account") String myAccount, Throwable error) {
        log.info("发生错误！！发生时间：" + System.currentTimeMillis() + "  用户是：" + myAccount);
        error.printStackTrace();
    }

    public static synchronized int getOnlineCount() {
        return onlineCount.get();
    }

    public static synchronized void addOnlineCount() {
        onlineCount.incrementAndGet();
    }

    public static synchronized void subOnlineCount() {
        onlineCount.decrementAndGet();
    }

    /**
     * 进行群发（所有在线用户可以收到）
     * @param message
     * @throws IOException
     */
    public void sendBatchToOnline(String message, @PathParam("account") String myAccount) throws IOException {
        Set<String> keyStrs = onlineSessionMap.keySet();
        for (String keyStr : keyStrs){
            Session toSession = onlineSessionMap.get(keyStr);
            sendMessageBySession(toSession, message);
            Integer fromUserId = userService.getUserByAccount(myAccount).getId();
            Integer toUserId = userService.getUserByAccount(keyStr).getId();
            saveMsg(message, fromUserId, toUserId, (byte)0);
        }
    }

    /**
     * 进行群发（所有用户包括离线可以收到）
     * @param message
     * @throws IOException
     */
    public void sendBatchToAll(String message, @PathParam("account") String myAccount) throws IOException {
        QueryWrapper<User> wrapper = new QueryWrapper<>();
        wrapper.select("account", "id");
        List<User> userList = userService.list(wrapper);
        User myUser = userService.getUserByAccount(myAccount);
        Integer toUserId = null;
        Set<String> keyStrs = onlineSessionMap.keySet();
        for (String keyStr : keyStrs) {
            Session toSession = onlineSessionMap.get(keyStr);
            sendMessageBySession(toSession, message);
            toUserId = userService.getUserByAccount(keyStr).getId();
            saveMsg(message, myUser.getId(), toUserId, (byte)0);
        }
        List<String> accountList = userList.stream().map(i -> i.getAccount()).collect(Collectors.toList());
        accountList.removeAll(keyStrs);
        for (String account : accountList){
            toUserId = userService.getUserByAccount(account).getId();
            saveMsg(message, myUser.getId(), toUserId, (byte)0);
        }
    }

    /**
     * 发送给指定用户账户
     * @param toAccount
     * @param message
     * @throws IOException
     */
    public void sendMessageByAccount(String toAccount, String message, String myAccount, Session session) throws IOException {
        Session toSession = onlineSessionMap.get(toAccount);
        User fromUser = userService.getUserByAccount(myAccount);
        User toUser = userService.getUserByAccount(toAccount);
        if (toUser == null){
            onClose(session, myAccount);
        }
        if (toSession == null){
            saveMsg(message, fromUser.getId(), toUser.getId(), (byte)0);
        } else {
            toSession.getBasicRemote().sendText(message);
            saveMsg(message, fromUser.getId(), toUser.getId(), (byte)0);
        }
    }

    /**
     * 发送给指定用户session
     * @param toSession
     * @param message
     * @throws IOException
     */
    public void sendMessageBySession(Session toSession, String message) throws IOException {
        toSession.getBasicRemote().sendText(message);
    }

    public void saveMsg(String message, Integer fromUserId, Integer toUserId, Byte isRead){
        Chat chat = new Chat();
        chat.setId(null);
        chat.setActive((byte)1);
        chat.setContent(message);
        chat.setFromUserId(fromUserId);
        chat.setToUserId(toUserId);
        chat.setIsRead(isRead);
        chatService.save(chat);
    }


}
